// Copyright 2012 Google Inc. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

// ---
// Simple way to start this server:
//   GLOG_logtostderr=1 GLOG_v=2 ./out/Default/seven_summits --tcp_port=10000

#include <assert.h>
#include <errno.h>
#include <netinet/tcp.h>
#include <sys/socket.h>
#include <fcntl.h>
#include <unistd.h>
#include <pthread.h>
#include <string.h>
#include <stdint.h>
#include <stdlib.h>
#include <stdio.h>
#include <sysexits.h>

#include <iostream>
#include <sstream>
#include <string>

#include "event2/buffer.h"
#include "event2/bufferevent.h"
#include "event2/event.h"
#include "event2/thread.h"
#include "glog/logging.h"
#include "llvm/Support/Signals.h"
#include "llvm/Support/CommandLine.h"
#include "llvm/Support/ManagedStatic.h"

#include "connection.h"
#include "connection_queue.h"
#include "worker.h"
#include "worker_pool.h"

// reference WorkerFactory::Registerer so linker does not strip it out
#include "stp_worker.h"

#define BACKLOG 1 << 10

namespace seven_summits {
namespace {

llvm::cl::opt<std::string> clWorker(
    "worker",
    llvm::cl::desc("which worker (i.e. solver implementation) to use"),
    llvm::cl::value_desc("worker class"),
    llvm::cl::init("StpWorker"));

llvm::cl::opt<int> clNumWorkers(
    "num_workers",
    llvm::cl::desc("number of concurrent workers"),
    llvm::cl::value_desc("int"),
    llvm::cl::init(1 << 10));

llvm::cl::opt<int> clNumConnections(
    "num_connections",
    llvm::cl::desc("max number of connections"),
    llvm::cl::value_desc("int"),
    llvm::cl::init(1 << 13));

llvm::cl::opt<int> clTcpPort(
    "tcp_port",
    llvm::cl::desc("TCP port number to listen on"),
    llvm::cl::value_desc("port"),
    llvm::cl::init(-1));

WorkerPool* worker_pool_;
ConnectionQueue connection_queue_;

void init_worker_pool() {
  worker_pool_ = new WorkerPool(clNumWorkers, WorkerFactory::GetByName(clWorker));
  worker_pool_->Init();
}

void init_connection_pool() {
  for (int i = 0; i < clNumConnections; ++i) {
    Connection* c = new Connection();
    c->set_pool(&connection_queue_);
    connection_queue_.Push(c);
  }
}

void do_accept(evutil_socket_t listener, short event, void *arg) {
  struct event_base *base = reinterpret_cast<event_base*>(arg);
  struct sockaddr_storage ss;
  socklen_t slen = sizeof(ss);
  int fd = accept(listener, (struct sockaddr*)&ss, &slen);
  if (fd < 0) { // TODO: eagain?
    LOG(ERROR) << "do_accept";
  } else if (fd > FD_SETSIZE) {
    evutil_closesocket(fd);
  } else {
    evutil_make_socket_nonblocking(fd);

    Connection* c = connection_queue_.Pop();
    Worker* w = worker_pool_->GetWorker();
    c->Init(base, fd, w);
  }
}

evutil_socket_t new_socket(struct addrinfo *ai) {
    int sfd;
    int flags;

    if ((sfd = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol)) == -1) {
        return -1;
    }

    return sfd;
}

// Create a socket and bind it to a specific port number.
// @param port the port number to bind to
// @param portnumber_file A filepointer to write the port numbers to
//        when they are successfully added to the list of ports we
//        listen on.
int server_socket(int port, struct event_base* base) {
    int sfd;
    struct linger ling = {0, 0};
    struct addrinfo *ai;
    struct addrinfo hints;
    hints.ai_flags = AI_PASSIVE;
    hints.ai_family = AF_UNSPEC;
    char port_buf[NI_MAXSERV];
    int error;
    int flags =1;

    hints.ai_socktype = SOCK_STREAM;
    hints.ai_protocol = 0;

    if (port == -1) {
        port = 0;
    }
    snprintf(port_buf, sizeof(port_buf), "%d", port);

    error = getaddrinfo(NULL, port_buf, &hints, &ai);
    if (error != 0) {
        if (error != EAI_SYSTEM)
          fprintf(stderr, "getaddrinfo(): %s\n", gai_strerror(error));
        else
          perror("getaddrinfo()");
        return 1;
    }

    for (struct addrinfo* next = ai; next; next = next->ai_next) {
      if ((sfd = new_socket(next)) == -1) {
        if (errno == EMFILE) {
          // we're out of fds
          perror("server_socket");
          exit(EX_OSERR);
        }
        continue;
      }

      if (next->ai_family == AF_INET6) {
        error = setsockopt(sfd, IPPROTO_IPV6, IPV6_V6ONLY, (char *) &flags, sizeof(flags));
        if (error != 0) {
          perror("setsockopt");
          close(sfd);
          continue;
        }
      }

      setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, (void *)&flags, sizeof(flags));
      error = setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, (void *)&flags, sizeof(flags));
      if (error != 0)
        perror("setsockopt");

      error = setsockopt(sfd, SOL_SOCKET, SO_LINGER, (void *)&ling, sizeof(ling));
      if (error != 0)
        perror("setsockopt");

      error = setsockopt(sfd, IPPROTO_TCP, TCP_NODELAY, (void *)&flags, sizeof(flags));
      if (error != 0)
        perror("setsockopt");

      evutil_make_socket_nonblocking(sfd);

      if (bind(sfd, next->ai_addr, next->ai_addrlen) == -1) {
        if (errno != EADDRINUSE) {
          perror("bind()");
          close(sfd);
          freeaddrinfo(ai);
          return 1;
        }
        close(sfd);
      }
      if (listen(sfd, BACKLOG) == -1) {
        perror("listen()");
        close(sfd);
        freeaddrinfo(ai);
        return 1;
      }

      struct event *listener_event =
          event_new(base, sfd, EV_READ|EV_PERSIST, do_accept, (void*)base);
      event_add(listener_event, NULL);
    }

    freeaddrinfo(ai);

    // No errors in starting up connections
    return 0;
}

}  // namespace anonymous
}  // namespace seven_summits

int main(int argc, char **argv) {
  setvbuf(stdout, NULL, _IONBF, 0);
  google::InitGoogleLogging(argv[0]);
  llvm::llvm_start_multithreaded();
  CHECK(!evthread_use_pthreads()) << "libevent cannot use pthreads";

  llvm::sys::PrintStackTraceOnErrorSignal();
  llvm::cl::ParseCommandLineOptions(argc, argv);
  seven_summits::init_worker_pool();
  seven_summits::init_connection_pool();

  struct event_base* base = event_base_new();
  CHECK(base) << "cannot get an event_base";

  evutil_socket_t listener =
      seven_summits::server_socket(seven_summits::clTcpPort, base);
  event_base_dispatch(base);

  llvm::llvm_shutdown();
  return 0;
}
